﻿#include <cstdio>
#include <string>
#include "log.h"
#include <iostream>
#include <fstream>
#include "config.h"
#include "mqttmgr.h"
#include <list>
#include <mutex>
#include <poll.h>
#include <thread>
#include <chrono>
//#include <zmqpp/zmqpp.hpp>
#include <sys/eventfd.h> 
#include "simple-socket/TCPSocket.h"
#include "simple-socket/UDPSocket.h"
#include "time_util.h"
#include "pc_socks.h"

/*socket列表及锁*/
std::mutex mtx_lst;
std::list<NET::TCPSocket::Handle> hd_lst{}; //新接受的tcp连接句柄列表；提交到工作线程进行读写

int accept_sem_ = 0; // TCP listen 服务事件，有新的连接请求接入

std::mutex mtx_msg;               //多线程互斥锁
DataFifo<uint8_t> msg_data_(2048); //数据缓冲区，用于保存从飞控接收到的数据。
bool exit_ = false;
#ifdef UDP
//保存udp远端参数、时间的结构体
typedef struct udp_sock_timestamp_
{
    std::string ip;
    uint16_t port = 0;
    std::chrono::steady_clock::time_point time;
} udp_sock_timestamp_t;
std::map<std::string, udp_sock_timestamp_t> udp_map;      //udp远端列表
#endif

std::map<int, std::shared_ptr<PcSocket>> sock_map_{}; //pc连接列表

int radio_out(const std::string& topic, const void* data, size_t len);

static int mqttcb(const char* topic, const void* data, int sz)
{
    radio_out(std::string(topic), data, sz);
    return 0;
}

/**
 * @brief 电台发送操作，从系统接收，发送数据到TCP连接
 * @param data
 * @param len
 * @return
*/
int radio_out(const std::string& topic, const void* data, size_t len)
{
    std::lock_guard<std::mutex> lck(mtx_msg);
    msg_data_.push_back(data, len);
#if PRINT
    printf("to pc  ");
    for (size_t i = 0; i < len; i++)
    {
        printf("%02x ", (int)((uint8_t*)data)[i]);
    }
    printf("\n");
#endif // PRINT

    eventfd_write(accept_sem_, 1);
    return 0;
}

/**
 * @brief tcp监听线程；
 * listen 端口并接受连接，然后把socket放入队列；死循环不会返回
 * @param port_ 监听端口
 */
void tcp_server_thread(uint16_t port_)
{
    try
    {
        NET::TCPSocket server; //服务器
        server.bind(port_);    //绑定端口
        server.listen();       //监听

        while (!exit_)
        {
            auto handle = server.accept(); //接受连接
            std::lock_guard<std::mutex> lck(mtx_lst); //加锁，防止多线程操作 hd_lst
            hd_lst.emplace_back(handle);    //添加接受的连接对象到 hd_lst
            eventfd_write(accept_sem_, 1);   //使用信号触发 tcp_rcv_thread 进行读写处理
        }
    }
    catch (const std::exception& oe)
    {
        printf("accept err.msg %s\n", oe.what());
        //未处理错误-退出线程
        exit_ = true;
        log_sys->info("tcp server port {} thread exit. err msg:{}", port_, oe.what());
    }
    return;
}

/**
 * @brief 负责接收TCP客户端数据
 * @return
*/
static int tcp_rcv_thread()
{ 
    // UDP端口，直接在工作线程中创建
#ifdef UDP
    NET::UDPSocket udp_sock; //udp服务端口
    udp_sock.bind(jouav::Config::GetInstance().GetUdpServerPort()); //绑定端口
#endif
    uint8_t pack[1024]; //临时缓冲区
    struct pollfd pfds[32];
    while (!exit_)
    {
        pfds[0].fd = accept_sem_; //接收tcp服务器的信号量事件；事件发生时代表有新的tcp连接加入
        pfds[0].events = POLLIN;
#ifdef UDP
        pfds[1].fd = udp_sock.nativeHandle(); //监听udp
        pfds[1].events = POLLIN;
        int cnt = 2;
#else
        int cnt = 1;
        
#endif
        const int socket_pos = cnt;
        //添加所有的TCP连接到poll中
        for (auto& var : sock_map_)
        {
            pfds[cnt].fd = var.second->sock.nativeHandle();
            pfds[cnt].events = POLLIN | POLLERR | POLLHUP | POLLNVAL;
            cnt++;
            if (cnt > 31)
                break;
        }

        auto num = poll(pfds, cnt, 100000); // poll等待事件

        //有事件发生
        if (num)
        {
            //处理TCP连接的接收事件
            for (auto i = socket_pos; i < cnt; i++)
            {
                if (!pfds[i].revents)
                    continue;
                auto iter = sock_map_.find(pfds[i].fd); // TCP对象迭代器
                if (iter == sock_map_.end())            //正常情况不会运行到end，只是防止越界
                    continue;
                auto var = iter->second;

                //处理输入事件
                if (pfds[i].revents & POLLIN)
                {
                    auto res = var->rcv_from_sock(); //自动读取数据

                    //读取数据失败，关闭连接
                    if (res < 1)
                    {
                        try
                        {
                            printf("%s:%d disconnected\r\n", iter->second->sock.getForeignAddress().c_str(),
                                iter->second->sock.getForeignPort());
                        }
                        catch (const std::exception&)
                        {
                        }
                        close(pfds[i].fd);
                        sock_map_.erase(iter); //从map中去处该连接
                        continue;             //继续运行其他TCP的接收
                    }
                   
                    int pack_len = 0;
                    //读取完该TCP的缓冲区数据
                    while (true)
                    {
                        pack_len = var->fifo.get_pack(pack, sizeof(pack)); //分析读取包结构

                        //读取到有效的数据
                        if (pack_len > 0)
                        {
                            //把接收到数据进行处理，比如转发出去
                            jouav::MqttMgr::GetInstance().Publish(
                                jouav::Config::GetInstance().GetMqttPubTopic(),
                                pack,
                                pack_len
                            );
                        }
                        else
                            break;
                    };
                }
                else if (pfds[i].revents & POLLERR)          //错误事件的句柄，直接删除
                {
                    close(pfds[i].fd);
                    sock_map_.erase(iter);
                }
                else if (pfds[i].revents & POLLHUP)
                { //套接字已经关闭
                    sock_map_.erase(iter);
                }
                else if (pfds[i].revents & POLLNVAL)
                {
                    close(pfds[i].fd);
                    sock_map_.erase(iter);
                }
            }

            // listen服务器的事件处理
            if (pfds[0].revents)
            {
                eventfd_t val;
                eventfd_read(accept_sem_, &val); //读取清空信号量
                //确认存在需要处理的 hd_lst
                if (!hd_lst.empty())
                {
                    std::lock_guard<std::mutex> lck(mtx_lst); //加锁
                    //循环处理 hd_lst 里的所有句柄
                    for (auto& var : hd_lst)
                    {
                        auto sockt = std::make_shared<PcSocket>(var);      //转换为智能指针
                        sock_map_[sockt->sock.nativeHandle()] = sockt; //添加到 sock_map
                        try
                        {
                            //打印日志，可以不需要使用try...catch
                            std::cout << "new socket in " << sockt->sock.getForeignAddress() << ":" << sockt->sock.getForeignPort() << std::endl;
                        }
                        catch (const std::exception&)
                        {
                        }
                    }
                    hd_lst.clear(); //处理完成，清空 hd_lst
                }
                else if (!msg_data_.empty())
                {
                    std::unique_lock<std::mutex> lck(mtx_msg);
                    auto res = msg_data_.pop_front_all(pack, sizeof(pack));
                    lck.unlock();
                    //给所有tcp客户端发送数据
                    for (auto& var : sock_map_)
                        var.second->sock.send(pack, res);

#ifdef UDP                   
                    //给所有udp客户端发送数据
                    auto tick = std::chrono::steady_clock::now();
                    for (auto iter = udp_map.begin(); iter != udp_map.end();)
                    {
                        if (tick - iter->second.time > std::chrono::milliseconds(20000)) //该sock已经超时，应该删除
                        {
                            printf("del udp socket: %s:%d", iter->second.ip.c_str(), iter->second.port);
                            iter = udp_map.erase(iter);
                            continue;
                        }
                        else
                            udp_sock.sendTo(pack, res, iter->second.ip, iter->second.port);//发送数据
                        ++iter;//不删除，应手动移动迭代器
                    }
#endif
                }
            }
#ifdef UDP
            if (pfds[1].revents) //接收udp数据
            {
                std::string ip;
                uint16_t port = 0;
                auto res = udp_sock.receiveFrom(pack, sizeof pack, ip, port); //读取数据长度
                if (res < 1)
                    continue;

                //把接收到数据进行处理，比如转发出去
                jouav::MqttMgr::GetInstance().Publish(
                    jouav::Config::GetInstance().GetMqttPubTopic(),
                    pack,
                    res
                );
                auto mask = std::to_string(port) + " " + ip; //构造字符串key用于map管理
                auto& var = udp_map[mask];              //取value引用
                if (!var.port)                          //port为0，代表此key之前不存在
                {
                    printf("new udp socket: %s:%d", var.ip.c_str(), var.port);
                    var.ip = ip;     //填入ip
                    var.port = port; //填入端口
                }
                var.time = std::chrono::steady_clock::now(); //标记时间
            }
#endif
        }
        else 
        {
            log_sys->info("tcp poll return. errno {}.", errno);
        }
    }
    log_sys->info("tcp rcv thread exit.");
    return 0;
}

int main(int argc, char* argv[])
{
    if (argc == 2 && argv[1] == std::string("-B"))
    {
        auto pid = fork();
        if (pid != 0) //父进程
        {
            std::fstream f_pid("/run/datalinksim-cloud.pid", std::ios::out);
            if (f_pid.is_open())
            {
                f_pid << std::to_string(pid);
                f_pid.close();
            }
            std::cout << "Main Progress exit" << std::endl;
            return 0;
        }
        std::cout << "Child Progress running" << std::endl;
    }

    sleep(2); //延时两秒等服务启动
    sock_map_.clear();
    //初始化配置文件
    if (argc >= 3 && argv[1] == std::string("-c"))
    {
        jouav::Config::GetInstance().Init(argv[2]);
    }
    else
    {
        jouav::Config::GetInstance().Init();
    }

    //初始化日志
    init_log(jouav::Config::GetInstance().GetLogPath());

    jouav::MqttMgr::GetInstance().SetSubCb(&mqttcb);

    int res = jouav::MqttMgr::GetInstance().Init();
    if (res)
    {
        return -1;
    }

    accept_sem_ = eventfd(0, 0); //创建信号量用于listen线程、工作线程同步
    //工作主线程，从UDP/TCP接收PC数据
    std::thread thd_task(tcp_rcv_thread);

    //监听线程，TCP服务器
    std::thread thd_listen(tcp_server_thread, jouav::Config::GetInstance().GetTcpServerPort());

    while (true)
    {
        sleep(1);
    }

    thd_task.join();
    thd_listen.join();
    printf("%s exit.\n", "datalinksim_clouds");
    return 0;
}